package com.greenhouse.protocol.channel;

import java.util.ArrayList;
import java.util.List;
import java.util.Optional;
import java.util.Set;
import java.util.stream.Collectors;

import org.apache.commons.lang3.StringUtils;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;

import com.greenhouse.protocol.BaseApi;
import com.greenhouse.protocol.BaseAuthService;
import com.greenhouse.protocol.ChannelService;
import com.greenhouse.protocol.common.enums.ConfirmStatus;
import com.greenhouse.protocol.mqtt.ServerMqttHandlerService;
import com.greenhouse.protocol.queue.MessageTransfer;

import io.netty.channel.Channel;
import io.netty.handler.codec.mqtt.MqttConnAckMessage;
import io.netty.handler.codec.mqtt.MqttConnAckVariableHeader;
import io.netty.handler.codec.mqtt.MqttConnectMessage;
import io.netty.handler.codec.mqtt.MqttConnectPayload;
import io.netty.handler.codec.mqtt.MqttConnectReturnCode;
import io.netty.handler.codec.mqtt.MqttFixedHeader;
import io.netty.handler.codec.mqtt.MqttMessage;
import io.netty.handler.codec.mqtt.MqttMessageIdVariableHeader;
import io.netty.handler.codec.mqtt.MqttMessageType;
import io.netty.handler.codec.mqtt.MqttPublishMessage;
import io.netty.handler.codec.mqtt.MqttQoS;
import io.netty.handler.codec.mqtt.MqttSubAckMessage;
import io.netty.handler.codec.mqtt.MqttSubAckPayload;
import io.netty.handler.codec.mqtt.MqttSubscribeMessage;
import io.netty.handler.codec.mqtt.MqttUnsubAckMessage;
import io.netty.handler.codec.mqtt.MqttUnsubscribeMessage;
import io.netty.handler.timeout.IdleStateEvent;
import lombok.extern.slf4j.Slf4j;

/**
 * 消息处理service
 *
 * @author lxr
 * @create 2017-11-21 11:13
 **/
@Component
@Slf4j
public class MqttHandlerService extends ServerMqttHandlerService implements BaseApi {

	@Autowired
	ChannelService mqttChannelService;

	private final MessageTransfer messageTransfer;

	private final BaseAuthService baseAuthService;

	public MqttHandlerService(MessageTransfer messageTransfer, BaseAuthService baseAuthService) {
		this.messageTransfer = messageTransfer;
		this.baseAuthService = baseAuthService;
	}

	/**
	 * 登录
	 *
	 */
	@Override
	public boolean login(Channel channel, MqttConnectMessage mqttConnectMessage) {
//        校验规则 自定义校验规则
		MqttConnectPayload payload = mqttConnectMessage.payload();
		String deviceId = payload.clientIdentifier();
		if (StringUtils.isBlank(deviceId)) {
			MqttConnectReturnCode connectReturnCode = MqttConnectReturnCode.CONNECTION_REFUSED_IDENTIFIER_REJECTED;
			connectBack(channel, connectReturnCode);
			return false;
		}

		if (mqttConnectMessage.variableHeader().hasPassword() && mqttConnectMessage.variableHeader().hasUserName()
				&& !baseAuthService.authorized(payload.userName(), payload.password())) {
			MqttConnectReturnCode connectReturnCode = MqttConnectReturnCode.CONNECTION_REFUSED_BAD_USER_NAME_OR_PASSWORD;
			connectBack(channel, connectReturnCode);
			return false;
		}
		return Optional.ofNullable(mqttChannelService.getMqttChannel(deviceId)).map(mqttChannel -> {
			switch (mqttChannel.getSessionStatus()) {
			case OPEN:
				return false;
			}
			mqttChannelService.loginSuccess(channel, deviceId, mqttConnectMessage);
			return true;
		}).orElseGet(() -> {
			mqttChannelService.loginSuccess(channel, deviceId, mqttConnectMessage);
			return true;
		});

	}

	private void connectBack(Channel channel, MqttConnectReturnCode connectReturnCode) {
		MqttConnAckVariableHeader mqttConnAckVariableHeader = new MqttConnAckVariableHeader(connectReturnCode, true);
		MqttFixedHeader mqttFixedHeader = new MqttFixedHeader(MqttMessageType.CONNACK, false, MqttQoS.AT_MOST_ONCE,
				false, 0x02);
		MqttConnAckMessage connAck = new MqttConnAckMessage(mqttFixedHeader, mqttConnAckVariableHeader);
		channel.writeAndFlush(connAck);
	}

	/**
	 * 发布
	 */
	@Override
	public void publish(Channel channel, MqttPublishMessage mqttPublishMessage) {
		mqttChannelService.publishSuccess(channel, mqttPublishMessage);
	}

	/**
	 * 订阅
	 */
	@Override
	public void subscribe(Channel channel, MqttSubscribeMessage mqttSubscribeMessage) {
		Set<String> topics = mqttSubscribeMessage.payload().topicSubscriptions().stream()
				.map(mqttTopicSubscription -> mqttTopicSubscription.topicName()).collect(Collectors.toSet());
		mqttChannelService.suscribeSuccess(mqttChannelService.getDeviceId(channel), topics);
		subBack(channel, mqttSubscribeMessage, topics.size());
	}

	private void subBack(Channel channel, MqttSubscribeMessage mqttSubscribeMessage, int num) {
		MqttFixedHeader mqttFixedHeader = new MqttFixedHeader(MqttMessageType.SUBACK, false, MqttQoS.AT_MOST_ONCE,
				false, 0);
		MqttMessageIdVariableHeader variableHeader = MqttMessageIdVariableHeader
				.from(mqttSubscribeMessage.variableHeader().messageId());
		List<Integer> grantedQoSLevels = new ArrayList<>(num);
		for (int i = 0; i < num; i++) {
			grantedQoSLevels.add(mqttSubscribeMessage.payload().topicSubscriptions().get(i).qualityOfService().value());
		}
		MqttSubAckPayload payload = new MqttSubAckPayload(grantedQoSLevels);
		MqttSubAckMessage mqttSubAckMessage = new MqttSubAckMessage(mqttFixedHeader, variableHeader, payload);
		channel.writeAndFlush(mqttSubAckMessage);
	}

	/**
	 * 关闭通道
	 */
	@Override
	public void close(Channel channel) {
		mqttChannelService.closeSuccess(mqttChannelService.getDeviceId(channel), false);
		channel.close();
	}

	/**
	 * 回复pong消息
	 */
	@Override
	public void pong(Channel channel) {
		if (channel.isOpen() && channel.isActive() && channel.isWritable()) {
			log.info("收到来自：【" + channel.remoteAddress().toString() + "】心跳");
			MqttFixedHeader fixedHeader = new MqttFixedHeader(MqttMessageType.PINGRESP, false, MqttQoS.AT_MOST_ONCE,
					false, 0);
			channel.writeAndFlush(new MqttMessage(fixedHeader));
		}
	}

	/**
	 * 取消订阅
	 */
	@Override
	public void unsubscribe(Channel channel, MqttUnsubscribeMessage mqttMessage) {
		List<String> topics1 = mqttMessage.payload().topics();
		mqttChannelService.unsubscribe(mqttChannelService.getDeviceId(channel), topics1);
		unSubBack(channel, mqttMessage.variableHeader().messageId());
	}

	/**
	 * 回复取消订阅
	 */
	private void unSubBack(Channel channel, int messageId) {
		MqttFixedHeader mqttFixedHeader = new MqttFixedHeader(MqttMessageType.UNSUBACK, false, MqttQoS.AT_MOST_ONCE,
				false, 0x02);
		MqttMessageIdVariableHeader variableHeader = MqttMessageIdVariableHeader.from(messageId);
		MqttUnsubAckMessage mqttUnsubAckMessage = new MqttUnsubAckMessage(mqttFixedHeader, variableHeader);
		channel.writeAndFlush(mqttUnsubAckMessage);
	}

	/**
	 * 消息回复确认(qos1 级别 保证收到消息  但是可能会重复)
	 */
	@Override
	public void puback(Channel channel, MqttMessage mqttMessage) {
		MqttMessageIdVariableHeader messageIdVariableHeader = (MqttMessageIdVariableHeader) mqttMessage
				.variableHeader();
		int messageId = messageIdVariableHeader.messageId();
		Optional.ofNullable(mqttChannelService.getMqttChannel(mqttChannelService.getDeviceId(channel))
				.getSendMqttMessage(messageId)).ifPresent(msg -> msg.setConfirmStatus(ConfirmStatus.COMPLETE)); // 复制为空
		messageTransfer.removeQueue(channel, messageId);
	}

	/**
	 * disconnect 主动断线
	 */
	@Override
	public void disconnect(Channel channel) {
		mqttChannelService.closeSuccess(mqttChannelService.getDeviceId(channel), true);
	}

	/**
	 * qos2 发布收到
	 */
	@Override
	public void pubrec(Channel channel, MqttMessage mqttMessage) {
		MqttMessageIdVariableHeader messageIdVariableHeader = (MqttMessageIdVariableHeader) mqttMessage
				.variableHeader();
		int messageId = messageIdVariableHeader.messageId();
		Optional.ofNullable(mqttChannelService.getMqttChannel(mqttChannelService.getDeviceId(channel))
				.getSendMqttMessage(messageId)).ifPresent(msg -> msg.setConfirmStatus(ConfirmStatus.PUBREL)); // 复制为空
		mqttChannelService.doPubrec(channel, messageId);
	}

	/**
	 * qos2 发布释放
	 */
	@Override
	public void pubrel(Channel channel, MqttMessage mqttMessage) {
		MqttMessageIdVariableHeader mqttMessageIdVariableHeader = (MqttMessageIdVariableHeader) mqttMessage
				.variableHeader();
		int messageId = mqttMessageIdVariableHeader.messageId();
		Optional.ofNullable(mqttChannelService.getMqttChannel(mqttChannelService.getDeviceId(channel))
				.getSendMqttMessage(messageId)).ifPresent(msg -> msg.setConfirmStatus(ConfirmStatus.COMPLETE));
		messageTransfer.removeQueue(channel, messageId);
		mqttChannelService.doPubrel(channel, messageId);
	}

	/**
	 * qos2 发布完成
	 */
	@Override
	public void pubcomp(Channel channel, MqttMessage mqttMessage) {
		MqttMessageIdVariableHeader mqttMessageIdVariableHeader = (MqttMessageIdVariableHeader) mqttMessage
				.variableHeader();
		int messageId = mqttMessageIdVariableHeader.messageId();
		Optional.ofNullable(mqttChannelService.getMqttChannel(mqttChannelService.getDeviceId(channel))
				.getSendMqttMessage(messageId)).ifPresent(msg -> msg.setConfirmStatus(ConfirmStatus.COMPLETE));
		messageTransfer.removeQueue(channel, messageId);
	}

	@Override
	public void doTimeOut(Channel channel, IdleStateEvent evt) {
		log.info("【PingPongService：doTimeOut 心跳超时】" + channel.remoteAddress() + "【channel 关闭】");
		switch (evt.state()) {
		case READER_IDLE:
			close(channel);
		case WRITER_IDLE:
			close(channel);
		case ALL_IDLE:
			close(channel);
		}
	}

}
